package com.haoziqi.chapter_01;

import com.sun.prism.shader.AlphaOne_LinearGradient_AlphaTest_Loader;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * TODO 批处理，处理文本数据
 * description
 * created by A on 2021/3/2
 */
public class BatchDemo {
    public static void main(String[] args) throws Exception {
        //1.创建执行环境
        ExecutionEnvironment env=ExecutionEnvironment.getExecutionEnvironment();
        //2.从文件读取数据 默认按行读取，（存储的元素就是每行的文本）
        DataSource<String> lines = env.readTextFile("input/word.txt");
        //3.转换数据格式
        FlatMapOperator<String, Tuple2<String, Long>> wordandOne= lines.flatMap((String line, Collector<Tuple2<String, Long>> out) -> {
            String[] s = line.split(" ");
            for (String word : s) {
                out.collect(Tuple2.of(word, 1L));
            }
        }).returns(Types.TUPLE(Types.STRING, Types.LONG));
        UnsortedGrouping<Tuple2<String, Long>> wordAndOneUG = wordandOne.groupBy(0);
        AggregateOperator<Tuple2<String, Long>> sum = wordAndOneUG.sum(1);
        sum.print();
    }
}
